package com.navinfo.opentsp.platform.computing.parallel.application;


import com.mongodb.spark.MongoSpark;
import com.mongodb.spark.config.ReadConfig;
import com.mongodb.spark.rdd.api.java.JavaMongoRDD;
import com.navinfo.opentsp.platform.computing.parallel.util.PropertiesUtil;

import com.navinfo.stream.qingqi.protocol.java.LCLocationData;
import com.navinfo.stream.qingqi.protocol.java.LCStatusType;
import com.navinfo.stream.qingqi.protocol.java.LCStatusType.StatusType;
import com.navinfo.stream.qingqi.protocol.java.LCVehicleStatusData;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.Partitioner;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.broadcast.Broadcast;
import org.bson.Document;
import org.bson.types.Binary;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.Tuple2;

import java.io.*;
import java.sql.*;
import java.text.SimpleDateFormat;
import java.util.*;


/**
 * Created by Administrator on 2018/1/4.
 */
public class ExportGpsLocationData {

    public static final LCStatusType.StatusType[] statusTypeArray = {
            StatusType.oilValue , StatusType.rotation, StatusType.coolingWaterTem, StatusType.totalFuelConsumption,
            StatusType.fuelConsumptionRate, StatusType.engineTorMode, StatusType.driverEnginePercentTor, StatusType.actualEnginePercentTor,
            StatusType.accPedalPos,StatusType.nominalFrictionPercentTrq,StatusType.parkingBrakeSwitch, StatusType.wheelBasedVehicleSpd,
            StatusType.cruiseCtrlActive, StatusType.brakeSwitch,  StatusType.clutchSwitch , StatusType.cruiseCtrlCoastSwitch,
            StatusType.cruiseCtrlResumeSwitch,StatusType.cruiseCtrlAccSwitch ,StatusType.cruiseCtrlSetSpd,StatusType.cruiseCtrlStates ,
            StatusType.outPutRoateSpeed
    };
    public static final String NEXT_LINE = java.security.AccessController.doPrivileged(new sun.security.action.GetPropertyAction("line.separator"));
    public static final int MAX_HEIGHT = 4000;
    private static Logger logger = LoggerFactory.getLogger(ExportGpsLocationData.class);
    public static void main(String[] args) {
        String day = "";
        String mode = null;
        String csvPath = "d:/";
        if(args == null || args.length == 0){
            day = "20180313";//DateUtils.format(new Date(), DateUtils.DateFormat.YYYYMMDD);
            mode = "local[*]";
        } else {
            day = args[0];
            csvPath = args[1];
        }
        Map<String,String> config = PropertiesUtil.getProperties();
        Map<String, List<String>> vehicleInfoCache = new HashMap<String, List<String>>();
        Map<String, String> commonDataCache = new HashMap<String, String>();
        commonDataCache.put("csvPath", csvPath);
        System.out.println(csvPath);
        loadVehicleInfo(vehicleInfoCache, commonDataCache);
        SparkConf conf = initConf(mode);
        JavaSparkContext jsc = null;

        try {

            long queryStart = System.currentTimeMillis();
            Map<String, String> readOverrides = new HashMap<String, String>();
            readOverrides.put("collection", "GpsDataEntity" + "_" + day.substring(2, 6));
            readOverrides.put("readPreference.name", "secondary");
            jsc = new JavaSparkContext(conf);
            ReadConfig readConfig = ReadConfig.create(jsc).withOptions(readOverrides);
            Document filter = Document.parse("{ $match: {day : '"+ day + "'} }");
            JavaMongoRDD<Document> rdd = MongoSpark.load(jsc, readConfig).withPipeline(Collections.singletonList(filter));
            long queryTotalTime = System.currentTimeMillis() - queryStart;
            System.out.println("load data from mongo, time=" + queryTotalTime);


            int partitionsNum = rdd.partitions().size();
            commonDataCache.put("partitionsNum", String.valueOf(partitionsNum));
            final Broadcast<Map<String, List<String>>> vehicleInfoMap = jsc.broadcast(vehicleInfoCache);
            final Broadcast<Map<String, String>> commonDataMap = jsc.broadcast(commonDataCache);

            rdd.mapPartitionsToPair(new PairFlatMapFunction<Iterator<Document>, String, List<String>>(){
                @Override
                public Iterable<Tuple2<String, List<String>>> call(Iterator<Document> documentIterator) throws Exception {
                    return handleDataToPair(documentIterator, vehicleInfoMap, commonDataMap);
                }
            }).partitionBy(new ExportDataPartitioner(Integer.parseInt(commonDataMap.getValue().get("partitionsNum"))))
                    .foreachPartition(new VoidFunction<Iterator<Tuple2<String, List<String>>>>() {
                        @Override
                        public void call(Iterator<Tuple2<String, List<String>>> tuple2Iterator) throws Exception {
                            while(tuple2Iterator.hasNext()){
                                Tuple2<String, List<String>> data = tuple2Iterator.next();
                                List<String> dataBeans = data._2;
                                String tidGpsDate = data._1;
                                String tid = tidGpsDate.split("_")[0];
                                String day = tidGpsDate.split("_")[1];
                                String fileName = commonDataMap.getValue().get(tid) + day;
                                String csvPath = commonDataMap.getValue().get("csvPath");
                                writeDataToCsv(dataBeans, csvPath, fileName + ".csv");
                            }
                        }
                    });
        }finally {
            jsc.close();
        }
    }

    private static void loadVehicleInfo(Map<String, List<String>> vehicleInfoCache, Map<String, String> commonDataCache) {

        commonDataCache.put("fieldUnitKey", PropertiesUtil.getProperties("fieldUnitKey"));
        commonDataCache.put("fieldUnitValue", PropertiesUtil.getProperties("fieldUnitValue"));
        commonDataCache.put("headerField", PropertiesUtil.getProperties("headerField"));

        String driver = "com.mysql.jdbc.Driver"; // URL指向要访问的数据库名game
        String url = PropertiesUtil.getProperties("mysql.url"); // MySQL配置时的用户名
        String user = PropertiesUtil.getProperties("mysql.username"); // MySQL配置时的密码
        String password = PropertiesUtil.getProperties("mysql.password");
        long startTime = System.currentTimeMillis();
        try (
                Connection conn = DriverManager.getConnection(url, user, password);
                PreparedStatement ps = conn.prepareStatement(PropertiesUtil.getProperties("vehicle.sql"));
                ResultSet rs = ps.executeQuery();
        ) {
            Class.forName(driver);
            ResultSetMetaData metaData = ps.getMetaData();
            long mysqlQueryTime = System.currentTimeMillis() - startTime;
            System.out.println("load data from mysql, time=" + mysqlQueryTime);
            while (rs.next()) {
                /*底盘号*/
                String chassisNum = getString(rs.getString(metaData.getColumnLabel(1)), "底盘号");
                String engineType = getString(rs.getString(metaData.getColumnLabel(2)), "发动机类型");
                String oriVin = rs.getString(metaData.getColumnLabel(3));
                String vin = getString(rs.getString(metaData.getColumnLabel(3)), "VIN");
                String driveForm = getString(rs.getString(metaData.getColumnLabel(4)), "驱动形式");
                String carType = getString(rs.getString(metaData.getColumnLabel(5)), "车辆类别");
                String platType = getString(rs.getString(metaData.getColumnLabel(6)), "平台车型");
                String productTime = getString(rs.getString(metaData.getColumnLabel(7)), "车辆生产日期");
                String teamName = getString(rs.getString(metaData.getColumnLabel(9)), "订单信息");
                String lhq = getString(rs.getString(metaData.getColumnLabel(10)), "离合器");
                String bsx = getString(rs.getString(metaData.getColumnLabel(11)), "变速箱类型");
                String hq = getString(rs.getString(metaData.getColumnLabel(12)),"后桥类型");
                String cj = getString(rs.getString(metaData.getColumnLabel(13)), "车架类型");
                String zj = getString(rs.getString(metaData.getColumnLabel(14)), "轴距");
                String qt = getString(rs.getString(metaData.getColumnLabel(15)), "其他");
                String lt = getString(rs.getString(metaData.getColumnLabel(16)), "轮胎类型");
                String tid = rs.getString(metaData.getColumnLabel(18));
                if(tid == null){
                    continue;
                }
                //设置车辆信息
                List<String> fields = new ArrayList<>();
                List<String> vehicleInfo = Arrays.asList(vin, chassisNum, platType, carType, engineType, lhq, driveForm, bsx, hq, cj, zj, lt, teamName, qt, productTime);
                fields.addAll(vehicleInfo);
                //tid与车辆信息绑定
                vehicleInfoCache.put(tid, fields);
                commonDataCache.put(tid, oriVin);
            }
        } catch (SQLException e) {
            logger.error("", e);
        } catch (ClassNotFoundException e) {
            logger.error("", e);
        }
    }

    private static List<Tuple2<String, List<String>>> handleDataToPair(Iterator<Document> documentIterator, Broadcast<Map<String, List<String>>> vehicleInfoCache, Broadcast<Map<String, String>> commonDataMap) {
        List<Tuple2<String, List<String>>> itorList = new ArrayList<>();
        Map<String, LastCacheInfo> lastCache = new HashMap();
        while(documentIterator.hasNext()){
            try{
                System.out.println("handleDataTo pair");
                Document data = documentIterator.next();
                if(data == null){
                    continue;
                }

                String tid = String.valueOf(data.getLong("tId"));
                String day = data.getString("day");
                String vin = commonDataMap.getValue().get(tid);
                String dataTime = "数据生成日期:" + day.substring(0, 4) + "/" + day.substring(4, 6) + "/" + day.substring(6, 8) + "/" + "00/00";
                List<Document> dataList = (List<Document>) data.get("dataList");
                List<String> beans = new ArrayList<String>();

                beans.addAll(vehicleInfoCache.getValue().get(tid));
                beans.add(dataTime);
                beans.add(commonDataMap.getValue().get("fieldUnitKey"));
                beans.add(commonDataMap.getValue().get("fieldUnitValue"));
                beans.add(commonDataMap.getValue().get("headerField"));
                SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy\\MM\\dd\\HH\\mm\\ss");
                for (Document doc : dataList) {
                    Binary binary = (Binary) doc.get("data");
                    LCLocationData.LocationData locationData = null;
                    try{
                        locationData = LCLocationData.LocationData.parseFrom(binary.getData());
                    }catch(Exception e){
                        logger.error("", e);
                        continue;
                    }
//                   if(locationData.getBatteryPower() != 0){
//                       continue;
//                   }

                    int status = getStatus(locationData);
                    int height = getHeight(lastCache, vin, locationData);
                    Map<StatusType, Object> statusDataMap = getStatusDataMap(locationData);
                    List<Object> fieldList = new ArrayList<>();
                    fieldList.add(dateFormat.format(Long.valueOf(locationData.getGpsDate() * 1000L)));
                    fieldList.add(locationData.getLongitude());
                    fieldList.add(locationData.getLatitude());
                    fieldList.add(height);
                    fieldList.add(locationData.getSpeed());
                    fieldList.add(status);
                    fieldList.add(statusDataMap.get(StatusType.mileage));
                    fieldList.add(locationData.getMileage()/1000.0D);
                    for(StatusType statusType : statusTypeArray){
                        fieldList.add(statusDataMap.get(statusType));
                    }
                    fieldList.add("0");
                    beans.add(StringUtils.join(fieldList, ","));
                }
                itorList.add(new Tuple2<String, List<String>>(tid + "_" + day, beans));
            } catch(Exception e){
                logger.error("", e);
            }
        }
        return itorList;
    }

    private static Map<StatusType, Object> getStatusDataMap(LCLocationData.LocationData locationData) {
        Map<StatusType, Object> statusDataMap = new HashMap<>();
        for (LCVehicleStatusData.VehicleStatusData vehStatus : locationData.getStatusAddition().getStatusList()) {
            if (vehStatus.getTypes() == StatusType.oilValue) {
                statusDataMap.put(StatusType.oilValue, vehStatus.getStatusValue() / 100L);
                continue;
            }
            if (vehStatus.getTypes() == StatusType.mileage) {
                statusDataMap.put(StatusType.mileage, vehStatus.getStatusValue() / 100.0D);
                continue;
            }
            if (vehStatus.getTypes() == StatusType.rotation) {
                statusDataMap.put(StatusType.rotation, vehStatus.getStatusValue() / 100L);
                continue;
            }
            if (vehStatus.getTypes() == StatusType.coolingWaterTem) {
                statusDataMap.put(StatusType.coolingWaterTem, vehStatus.getStatusValue() / 100L);
                continue;
            }
            if (vehStatus.getTypes() == StatusType.totalFuelConsumption) {
                statusDataMap.put(StatusType.totalFuelConsumption, vehStatus.getStatusValue() / 100L);
                continue;
            }
            if (vehStatus.getTypes() == StatusType.fuelConsumptionRate) {
                statusDataMap.put(StatusType.fuelConsumptionRate, (float)vehStatus.getStatusValue() / 100.0F / 1000.0F);
                continue;
            }
            if (vehStatus.getTypes() == StatusType.driverEnginePercentTor) {
                statusDataMap.put(StatusType.driverEnginePercentTor, vehStatus.getStatusValue() / 100L);
                continue;
            }
            if (vehStatus.getTypes() == StatusType.engineTorMode) {
                statusDataMap.put(StatusType.engineTorMode, vehStatus.getStatusValue() / 100L);
                continue;
            }
            if (vehStatus.getTypes() == StatusType.actualEnginePercentTor) {
                statusDataMap.put(StatusType.actualEnginePercentTor, vehStatus.getStatusValue() / 100L);
                continue;
            }
            if (vehStatus.getTypes() == StatusType.accPedalPos) {
                statusDataMap.put(StatusType.accPedalPos, vehStatus.getStatusValue() / 100L);
                continue;
            }
            if (vehStatus.getTypes() == StatusType.parkingBrakeSwitch) {
                statusDataMap.put(StatusType.parkingBrakeSwitch, vehStatus.getStatusValue() / 100L);
                continue;
            }
            if (vehStatus.getTypes() == StatusType.wheelBasedVehicleSpd) {
                statusDataMap.put(StatusType.wheelBasedVehicleSpd, vehStatus.getStatusValue() / 100L);
                continue;
            }
            if (vehStatus.getTypes() == StatusType.cruiseCtrlActive) {
                statusDataMap.put(StatusType.cruiseCtrlActive, vehStatus.getStatusValue() / 100L);
                continue;
            }
            if (vehStatus.getTypes() == StatusType.brakeSwitch) {
                statusDataMap.put(StatusType.brakeSwitch, vehStatus.getStatusValue() / 100L);
                continue;
            }
            if (vehStatus.getTypes() == StatusType.clutchSwitch) {
                statusDataMap.put(StatusType.clutchSwitch, vehStatus.getStatusValue() / 100L);
                continue;
            }
            if (vehStatus.getTypes() == StatusType.cruiseCtrlCoastSwitch) {
                statusDataMap.put(StatusType.cruiseCtrlCoastSwitch, vehStatus.getStatusValue() / 100L);
                continue;
            }
            if (vehStatus.getTypes() == StatusType.cruiseCtrlResumeSwitch) {
                statusDataMap.put(StatusType.cruiseCtrlResumeSwitch, vehStatus.getStatusValue() / 100L);
                continue;
            }
            if (vehStatus.getTypes() == StatusType.cruiseCtrlAccSwitch) {
                statusDataMap.put(StatusType.cruiseCtrlAccSwitch, vehStatus.getStatusValue() / 100L);
                continue;
            }
            if (vehStatus.getTypes() == StatusType.cruiseCtrlSetSpd) {
                statusDataMap.put(StatusType.cruiseCtrlSetSpd, vehStatus.getStatusValue() / 100L);
                continue;
            }
            if (vehStatus.getTypes() == StatusType.cruiseCtrlStates) {
                statusDataMap.put(StatusType.cruiseCtrlStates, vehStatus.getStatusValue() / 100L);
                continue;
            }
            if (vehStatus.getTypes() == StatusType.outPutRoateSpeed) {
                statusDataMap.put(StatusType.outPutRoateSpeed, vehStatus.getStatusValue() / 100L);
                continue;
            }
            if (vehStatus.getTypes() == StatusType.nominalFrictionPercentTrq) {
                statusDataMap.put(StatusType.nominalFrictionPercentTrq, vehStatus.getStatusValue() / 100L);
                continue;
            }
        }
        return statusDataMap;
    }

    private static int getStatus(LCLocationData.LocationData locationData) {
        int status = 0;
        if ((locationData.getStatus() & 1L) == 1L) {
            status = 1;
        }
        return status;
    }

    private static int getHeight(Map<String, LastCacheInfo> lastCache, String vin, LCLocationData.LocationData locationData) {
        int height = 0;
        if (!lastCache.containsKey(vin))
        {
            LastCacheInfo lastInfo = new LastCacheInfo();
            if (locationData.getHeight() > MAX_HEIGHT) {
                lastInfo.setHeight(500);
            } else {
                lastInfo.setHeight(locationData.getHeight());
            }
            lastInfo.setLat(locationData.getLatitude());
            lastInfo.setLng(locationData.getLongitude());
            lastCache.put(vin, lastInfo);
        } else {
            LastCacheInfo lastInfo = lastCache.get(vin);
            boolean flag = validateDistance(lastInfo, locationData.getLatitude(), locationData.getLongitude(), locationData.getHeight());
            if (!flag){
                height = lastInfo.getHeight();
            } else {
                height = locationData.getHeight();
                lastInfo.setHeight(locationData.getHeight());
            }
            lastInfo.setLat(locationData.getLatitude());
            lastInfo.setLng(locationData.getLongitude());
        }
        return height;
    }

    private static SparkConf initConf(String mode) {
        SparkConf conf = new SparkConf()
                .setAppName("ExportGpsLocationData")
                .set("spark.mongodb.input.uri",  PropertiesUtil.getProperties("spark.mongodb.input.uri"))//mongodb input 连接
                .set("spark.mongodb.input.partitioner", "MongoPaginateBySizePartitioner")
                .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");

        if(StringUtils.isNotBlank(mode)){
            conf.setMaster(mode);
        }
        return conf;
    }

    private static void writeDataToCsv(List<String> dataList, String path, String fileName){
        if(dataList == null){
            return;
        }
        File file = new File(path);
        if(!file.exists()){
            file.mkdirs();
        }
        BufferedWriter writer;
        try {
            writer = new BufferedWriter(new FileWriter(path + File.separator + fileName));
            for(String lineData : dataList){
                writer.write(lineData + NEXT_LINE);
            }
            if (writer != null) {
                writer.flush();
                writer.close();
            }
        } catch (IOException e) {
            logger.error("", e);
        }
    }

    private static String getString(String data, String prefix){
        return prefix + ":" + (data == null ? "" : data);
    }

    private static double EARTH_RADIUS = 6378.137D;

    private static double rad(double d)
    {
        return d * 3.141592653589793D / 180.0D;
    }

    private static double getDistance(double lat1, double lng1, double lat2, double lng2)
    {
        double radLat1 = rad(lat1);
        double radLat2 = rad(lat2);
        double a = Math.abs(radLat1 - radLat2);
        double b = Math.abs(rad(lng1) - rad(lng2));

        double s = 2.0D * Math.asin(Math.sqrt(Math.pow(Math.sin(a / 2.0D), 2.0D) +
                Math.cos(radLat1) * Math.cos(radLat2) *
                        Math.pow(Math.sin(b / 2.0D), 2.0D)));
        s *= EARTH_RADIUS;
        return Math.round(s * 1000.0D);
    }

    private static boolean validateDistance(LastCacheInfo info, long lat, long lng, int height)
    {
        double distance = getDistance(info.getLat(), info.getLng(), lat, lng);
        int ht = Math.abs(info.getHeight() - height);
        if (distance >= 2 * ht) {
            return true;
        }
        return false;
    }

    public static class LastCacheInfo implements Serializable {
        private long lat;
        private long lng;
        private int height;

        public long getLat()
        {
            return this.lat;
        }

        public void setLat(long lat)
        {
            this.lat = lat;
        }

        public long getLng()
        {
            return this.lng;
        }

        public void setLng(long lng)
        {
            this.lng = lng;
        }

        public int getHeight()
        {
            return this.height;
        }

        public void setHeight(int height)
        {
            this.height = height;
        }
    }

    public static class ExportDataPartitioner extends Partitioner {

        public ExportDataPartitioner(){}

        public ExportDataPartitioner(int partitionsNum){
            this.partitionsNum = partitionsNum;
        }

        private int partitionsNum;

        @Override
        public int numPartitions() {
            return partitionsNum;
        }

        @Override
        public int getPartition(Object o) {
            String key = String.valueOf(o);
            return (int)(Long.parseLong(key.split("_")[0])%numPartitions());
        }
    }
}
